

namespace Stee.CAP8.MessageBus.Kafka.Subscriber
{
    using System;
    using System.Threading.Tasks;
    using Confluent.Kafka;
    using NLog;
    using Polly;
    using Stee.CAP8.MessageBus.Interfaces.Delegates;
    using Stee.CAP8.MessageBus.Interfaces.Enums;
    using Stee.CAP8.MessageBus.Interfaces.Interfaces;
    using Stee.CAP8.MessageBus.Interfaces.Models;
    using Stee.CAP8.MessageBus.Kafka.Utility;

    /* 
     * COMMAND Messages
     * =====
     * Individial services can be designated with a group ID solely for command type messsages.
     * For notification service, it could be "notificationsvc", where messages to this topic will be consumed
     * only by notification service. 
     * Example command group IDs:
	 *     - notificationsvc
	 *     - dispatchsvc
	 *     - gissvc
     *
     *
     * EVENT Messages
     * =====
     * Based on events, if 2 services share the same group ID, only 1 service will process a specific message.
     * This is not ideal if the event was "unitlocation_update" where resource and GIS services require this message.
     * Therefore to allow both services access to an event message, 2 group IDs must be created.
     * Example event group IDs:
     *     - unit_create_{0}: notificationsvc, resourcesvc
     *     - unit_update_{0}: notificationsvc, resourcesvc
     *     - Eg.: unit_update_notificationsvc
     *
     *     - incident_create_{0}: notificationsvc, resourcesvc
     *     - incident_update_{0}: notificationsvc, resourcesvc
     *     - Eg.: incident_update_notificationsvc
     *
     *     - unitlocation_update_{0}: notificationsvc, dispatchsvc, gissvc
     *     - incidentlocation_update_{0}: notificationsvc, dispatchsvc, gissvc
     *     - Eg.: incidentlocation_update_notificationsvc
     *
     * Thereafter, command & event messages can be viewed as similar implementation.
     * 
     *
     * CONSUMER
     * =====
     * Kafka consumer uses polling to 'listen' to messages, therefore is not automatically made known to new messages.
     * Using a polling method, the consumer has a while loop to constantly poll for new messages(never ending).
     * On subscription, a polling task has to be created and maintained preferrably by the subscriber themselves.
     * The polling task will have a timer to check kafka consumer state.
     * If the consumer is 'stopped', the timer will trigger an event to restart message polling
     * 
     * Required information: (Topic, notification handler). Configuration is assumed to be known.
     * Logic: A recoverable task constantly polling for messages with subscribe as start, unsubscribe as stop.
     * 
     * 
     * SERIALIZATION / DESERIALIZATION Issue
     * =====
     * Kafka is able to publish messages of any serializable data type. The serialized message in byte[] will be published to
     * the broker. Since there is no limitation to the number of allowed data types per Kafka topic, a topic can potentially
     * contain messages of ANY data type. As a publisher, this will not be an issue as the type is known during
     * publishing (serialization).
     * 
     * As a consumer, a logic is needed to handle multiple data types even if it consumes only 1 topic. To complicate
     * things, the consumed message will need de-serialization which requires the data type information.
     * The solution to this is to save the data type with the published message. During consumption, the data type can be
     * extracted and with proper mapping (logic), de-serialized.
     * In the publisher implementation, data type will be saved as the last message header with constant key, "MessageType".
     * Consumer will then use the same key to retrieve header for the data type.
     * https://github.com/confluentinc/confluent-kafka-dotnet/issues/1183
     * 
     */
    /// <summary>
    /// Implements the subscriber/consumer for receiving messages. 
    /// Each subscriber will handle a specific messsage data type. 
    /// https://github.com/confluentinc/confluent-kafka-dotnet/blob/master/examples/Consumer/Program.cs
    /// </summary>
    public class SubscriberImpl : ISubscriber
    {
        #region Constant & Static Variables

        private readonly static Logger Logger = LogManager.GetCurrentClassLogger();


        /// <summary>
        /// Polly retry delay for policy
        /// </summary>
        private const int ConnectionRetryDelayMS = 2000;

        /// <summary>
        /// Timer interval to check polling state
        /// </summary>
        protected const int TimerIntervalSubscribe = 30000;

        #endregion Constant & Static Variables


        #region Fields

        /// <summary>
        /// Polling implmentation
        /// </summary>
        private PollingBase pollingImpl;

        /// <summary>
        /// Asynchronous operation for polling consumer messages
        /// </summary>
        private Task pollingTask;

        /// <summary>
        /// Flag indicating Polly retry connect logic is in progress
        /// </summary>
        private bool isPollyReconnect = false;

        /// <summary>
        /// Timer for re-subscribing
        /// </summary>
        protected readonly System.Timers.Timer ResubscribeTimer;

        /// <summary>
        /// Mutex lock
        /// </summary>
        private readonly object mutex = new object();

        /// <summary>
        /// Mutex lock
        /// </summary>
        private readonly object mutex2 = new object();

        #endregion Fields


        #region Properties

        /// <summary>
        /// Gets the message topic
        /// Mandatory configuration.
        /// </summary>
        public TopicName TopicName => this.pollingImpl?.TopicName;

        /// <summary>
        /// Gets the client group id string.
        /// All clients sharing the same group.id belong to the same group.
        /// Mandatory configuration.
        /// </summary>
        public string GroupID => this.pollingImpl?.GroupID;

        /// <summary>
        /// Gets an optional identifier of a Kafka consumer(in a consumer group) that is passed to a Kafka broker with every request.
        /// The sole purpose of this is to be able to track the source of requests beyond just ip and port
        /// by allowing a logical application name to be included in Kafka logs and monitoring aggregates.
        /// </summary>
        public string ClientID => this.pollingImpl?.ClientID;

        /// <summary>
        /// Gets the flag indicating that subscriber is initialized and ready for reading messages
        /// </summary>
        public bool IsInitialized { get; private set; } = false;

        /// <summary>
        /// Gets the flag indicating if consumer is currently consuming messages
        /// </summary>
        public bool IsConnected => this.pollingImpl?.IsPolling == true;

        /// <summary>
        /// Gets the flag indicating if publisher is disposed
        /// </summary>
        public bool IsDisposed => (this.pollingImpl == null) || this.pollingImpl.IsDisposed;

        #endregion Properties


        #region Constructors & Initializers

        /// <summary>
        /// Initializes a new instance of the SubscriberImpl class
        /// </summary>
        public SubscriberImpl()
        {
            //// Create re-subscribe timer
            this.ResubscribeTimer = new System.Timers.Timer()
            {
                AutoReset = true,
                Interval = TimerIntervalSubscribe,
            };
            this.ResubscribeTimer.Elapsed += this.ResubscribeTimerOnElapsed;
        }

        #endregion Constructors & Initializers


        #region Interface Implementation

        /// <summary>
        /// Initializes the subscription
        /// </summary>
        /// <param name="brokerConfigurationJSON">JSON configuration path</param>
        /// <param name="groupID">Subscriber group ID</param>
        /// <param name="topicName">Subscription topic</param>
        /// <param name="brokerErrorHandler">Broker error handler</param>
        /// <returns>Return TRUE if initialized else FALSE</returns>
        public bool InitializeGeneric(string brokerConfigurationJSON, string groupID, TopicName topicName, HandleBrokerError brokerErrorHandler)
        {
            try
            {
                if (this.IsInitialized)
                {
                    throw new InvalidOperationException($"Subscriber is already initialized. IsInit: {this.IsInitialized}");
                }

                ValidateHelper.ValidateTopicName(topicName);
                this.pollingImpl = new PollingGenericImpl(brokerConfigurationJSON, groupID, brokerErrorHandler)
                {
                    HandleError = this.HandleError,
                };

                return this.Initialize(topicName);
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Initializes the subscription
        /// </summary>
        /// <param name="brokerConfigurationJSON">JSON configuration path</param>
        /// <param name="groupID">Subscriber group ID</param>
        /// <param name="topicName">Subscription topic</param>
        /// <param name="brokerErrorHandler">Broker error handler</param>
        /// <returns>Return TRUE if initialized else FALSE</returns>
        public bool InitializeEvent(string brokerConfigurationJSON, string groupID, TopicName topicName, HandleBrokerError brokerErrorHandler)
        {
            try
            {
                if (this.IsInitialized)
                {
                    throw new InvalidOperationException($"Subscriber is already initialized. IsInit: {this.IsInitialized}");
                }

                ValidateHelper.ValidateTopicName(topicName);
                this.pollingImpl = new PollingEventImpl(brokerConfigurationJSON, groupID, brokerErrorHandler)
                {
                    HandleError = this.HandleError,
                };

                return this.Initialize(topicName);
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Initializes the subscription
        /// </summary>
        /// <param name="brokerConfigurationJSON">JSON configuration path</param>
        /// <param name="groupID">Subscriber group ID</param>
        /// <param name="topicName">Subscription topic</param>
        /// <param name="brokerErrorHandler">Broker error handler</param>
        /// <returns>Return TRUE if initialized else FALSE</returns>
        public bool InitializeCommand(string brokerConfigurationJSON, string groupID, TopicName topicName, HandleBrokerError brokerErrorHandler)
        {
            try
            {
                if (this.IsInitialized)
                {
                    throw new InvalidOperationException($"Subscriber is already initialized. IsInit: {this.IsInitialized}");
                }

                ValidateHelper.ValidateTopicName(topicName);
                this.pollingImpl = new PollingCommandImpl(brokerConfigurationJSON, groupID, brokerErrorHandler)
                {
                    HandleError = this.HandleError,
                };

                return this.Initialize(topicName);
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Subscribe to configured topic
        /// </summary>
        /// <param name="subscribeMessageHandler">Subscribe message handler</param>
        public void Subscribe(HandleSubscribe<GenericMessage> subscribeMessageHandler)
        {
            try
            {
                if (this.pollingImpl is PollingGenericImpl)
                {
                    var pollImpl = this.pollingImpl as PollingGenericImpl;
                    pollImpl.SubscribeHandler.AddHandler(
                        subscribeMessageHandler, this.TopicName.Name);
                }
                else
                {
                    throw new InvalidOperationException("Subscriber is not of generic type, unable to subscribe with generic handler.");
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Subscribe to configured topic
        /// </summary>
        /// <param name="subscribeEventHandler">Subscribe event handler</param>
        /// <param name="handlerPriority">Subscribe handler priority</param>
        /// <param name="isHandlerAsync">Flag indicating handler to process asynchronously</param>
        public void Subscribe(HandleSubscribe<EventMessage> subscribeEventHandler, PriorityLevel handlerPriority, bool isHandlerAsync)
        {
            try
            {
                if (this.pollingImpl is PollingEventImpl)
                {
                    var pollImpl = this.pollingImpl as PollingEventImpl;
                    pollImpl.SubscribeHandler.AddPrioritisedHandler(
                        subscribeEventHandler, handlerPriority, isHandlerAsync);
                }
                else
                {
                    throw new InvalidOperationException("Subscriber is not of event type, unable to subscribe with event handler.");
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Subscribe to configured topic
        /// </summary>
        /// <param name="subscribeCommandHandler">Subscribe command handler</param>
        public void Subscribe(HandleSubscribe<CommandMessage> subscribeCommandHandler)
        {
            try
            {
                if (this.pollingImpl is PollingCommandImpl)
                {
                    var pollImpl = this.pollingImpl as PollingCommandImpl;
                    pollImpl.SubscribeHandler.AddHandler(
                        subscribeCommandHandler, this.TopicName.Name);
                }
                else
                {
                    throw new InvalidOperationException("Subscriber is not of command type, unable to subscribe with command handler.");
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Subscribe to configured topic
        /// </summary>
        /// <param name="subscribeCommandHandler">Subscribe command handler</param>
        /// <param name="actionName"></param>
        public void Subscribe(HandleSubscribe<CommandMessage> subscribeCommandHandler, string actionName)
        {
            try
            {
                if (this.pollingImpl is PollingCommandImpl)
                {
                    var pollImpl = this.pollingImpl as PollingCommandImpl;
                    pollImpl.SubscribeHandler.AddActionHandler(
                        subscribeCommandHandler, this.TopicName.Name, actionName);
                }
                else
                {
                    throw new InvalidOperationException("Subscriber is not of command type, unable to subscribe with command handler.");
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Unsubscribe from configured topic
        /// </summary>
        /// <param name="subscribeMessageHandler">Subscribe message handler</param>
        public void Unsubscribe(HandleSubscribe<GenericMessage> subscribeMessageHandler)
        {
            try
            {
                if (this.pollingImpl is PollingGenericImpl)
                {
                    var pollImpl = this.pollingImpl as PollingGenericImpl;
                    pollImpl.SubscribeHandler.RemoveHandler(subscribeMessageHandler, this.TopicName.Name);
                }
                else
                {
                    throw new InvalidOperationException("Subscriber is not of generic type, unable to remove generic handler.");
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Unsubscribe from configured topic
        /// </summary>
        /// <param name="subscribeEventHandler">Subscribe event handler</param>
        /// <param name="handlerPriority">Subscribe handler priority</param>
        /// <param name="isHandlerAsync">Flag indicating handler to process asynchronously</param>
        public void Unsubscribe(HandleSubscribe<EventMessage> subscribeEventHandler, PriorityLevel handlerPriority, bool isHandlerAsync)
        {
            try
            {
                if (this.pollingImpl is PollingEventImpl)
                {
                    var pollImpl = this.pollingImpl as PollingEventImpl;
                    pollImpl.SubscribeHandler.RemovePrioritisedHandler(subscribeEventHandler, handlerPriority, isHandlerAsync);
                }
                else
                {
                    throw new InvalidOperationException("Subscriber is not of event type, unable to remove event handler.");
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Unsubscribe from configured topic
        /// </summary>
        /// <param name="subscribeCommandHandler">Subscribed command handler</param>
        public void Unsubscribe(HandleSubscribe<CommandMessage> subscribeCommandHandler)
        {
            try
            {
                if (this.pollingImpl is PollingCommandImpl)
                {
                    var pollImpl = this.pollingImpl as PollingCommandImpl;
                    pollImpl.SubscribeHandler.RemoveHandler(subscribeCommandHandler, this.TopicName.Name);
                }
                else
                {
                    throw new InvalidOperationException("Subscriber is not of command type, unable to remove command handler.");
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Unsubscribe from configured topic
        /// </summary>
        /// <param name="subscribeCommandHandler">Subscribed command handler</param>
        /// <param name="actionName">Action name</param>
        public void Unsubscribe(HandleSubscribe<CommandMessage> subscribeCommandHandler, string actionName)
        {
            try
            {
                if (this.pollingImpl is PollingCommandImpl)
                {
                    var pollImpl = this.pollingImpl as PollingCommandImpl;
                    pollImpl.SubscribeHandler.RemoveActionHandler(subscribeCommandHandler, this.TopicName.Name, actionName);
                }
                else
                {
                    throw new InvalidOperationException("Subscriber is not of command type, unable to remove command handler.");
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Commit offset of consumed message
        /// </summary>
        /// <param name="brokerMessage">Message consumed from Kafka cluster</param>
        public bool Commit(object brokerMessage)
        {
            return this.pollingImpl.Commit(brokerMessage);
        }

        /// <summary>
        /// Reset consumer group offset to latest
        /// </summary>
        /// <param name="topicName">Subscription topic</param>
        /// <returns>Return TRUE if the reset is successful else FALSE</returns>
        private bool ResetOffsetToLatest(TopicName topicName)
        {
            return this.pollingImpl.ResetOffsetToLatest(topicName);
        }

        /// <summary>
        /// Dispose subscriber
        /// </summary>
        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        #endregion Interface Implementation


        #region Private Methods

        /// <summary>
        /// Initializes the subscription
        /// </summary>
        /// <param name="topicName">Subscription topic</param>
        /// <returns>Return TRUE if initialized else FALSE</returns>
        private bool Initialize(TopicName topicName)
        {
            this.pollingImpl.Initialize(topicName);
            this.StartPoll();

            return this.IsInitialized = true;
        }

        /// <summary>
        /// Start poll task
        /// </summary>
        private void StartPoll()
        {
            try
            {
                lock (this.mutex)
                {
                    if (this.pollingTask == null)
                    {
                        this.pollingTask = Task.Run(() => this.pollingImpl.StartPolling());
                        this.pollingTask.ConfigureAwait(false);

                        this.RestartSubscribeTimer();
                    }
                    else
                    {
                        Logger.Warn("Polling task has already started.");
                    }
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Unsubscribe from topic
        /// </summary>
        private void StopPoll()
        {
            try
            {
                lock (this.mutex)
                {
                    this.StopSubscribeTimer();

                    //// Clear polling task
                    if (this.pollingTask != null && (this.pollingTask.IsCanceled || this.pollingTask.IsCompleted || this.pollingTask.IsFaulted))
                    {
                        this.pollingTask.Dispose();
                    }

                    this.pollingTask = null;
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Retry connecting polling consumer to Kafka broker
        /// </summary>
        private void RetryStartPolling()
        {
            if (this.isPollyReconnect)
            {
                Logger.Info("Retry to start polling in progress...");
                return;
            }

            lock (this.mutex2)
            {
                try
                {
                    this.isPollyReconnect = true;

                    Policy.HandleResult<bool>((isPolling) => !isPolling)
                        .WaitAndRetryForever(retryattempt => TimeSpan.FromSeconds(ConnectionRetryDelayMS))
                        .Execute(() =>
                        {
                            if (this.pollingImpl.IsPolling)
                            {
                                Logger.Info("Consumer polling started.");
                                return false;
                            }
                            else
                            {
                                Logger.Warn("Consumer polling stopped, attempt to restart polling...");
                                this.StopPoll();
                                this.StartPoll();

                                return true;
                            }
                        });
                }
                catch (Exception ex)
                {
                    Logger.Fatal(ex, ex.Message);
                }
                finally
                {
                    this.isPollyReconnect = false;
                }
            }
        }

        /// <summary>
        /// Consumer connection error handler
        /// </summary>
        /// <param name="error">Consumer error</param>
        private void HandleError(Error error)
        {
            //// If there are any error, a re-connect will start
            //// Note: This is a simple re-connect. More serious errors will need user intervention.
            if (error != null && error.Code != ErrorCode.NoError)
            {
                this.RetryStartPolling();
            }
        }

        /// <summary>
        /// Dispose
        /// </summary>
        /// <param name="disposing">If disposing equals true, the method has been called directly
        /// or indirectly by a user's code. Managed and unmanaged resources can be disposed.
        /// If disposing equals false, the method has been called by the runtime from inside the
        /// finalizer and you should not reference other objects. Only unmanaged resources can be disposed.</param>
        protected virtual void Dispose(bool disposing)
        {
            try
            {
                if (!this.IsDisposed)
                {
                    this.StopPoll();
                    this.pollingImpl?.Dispose();
                    this.pollingImpl = null;
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }


        #region Resubscribe Timer

        /// <summary>
        /// Start re-subscribe timer
        /// </summary>
        /// <returns>Return TRUE if started else FALSE</returns>
        private void RestartSubscribeTimer()
        {
            try
            {
                this.ResubscribeTimer.Stop();
                this.ResubscribeTimer.Start();
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
            }
        }

        /// <summary>
        /// Stop re-subscribe timer
        /// </summary>
        /// <returns>Return TRUE if stopped else FALSE</returns>
        private void StopSubscribeTimer()
        {
            try
            {
                this.ResubscribeTimer.Stop();
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
            }
        }

        /// <summary>
        /// Re-subscribe timer elapsed event
        /// </summary>
        /// <param name="sender">Trigger source</param>
        /// <param name="e">Event argument</param>
        private void ResubscribeTimerOnElapsed(object sender, System.Timers.ElapsedEventArgs e)
        {
            try
            {
                //// Log timer elapsed event
                Logger.Info("Restarting polling task...");

                if (!this.pollingImpl.IsCancelled && !this.pollingImpl.IsPolling)
                {
                    Logger.Warn("Polling task is not running, retry connection...");
                    this.RetryStartPolling();
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
            }
        }

        #endregion Resubscribe Timer

        #endregion Private Methods
    }
}
